package com.ruoyi.server.api.websocket;

import cn.hutool.json.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.server.api.message.entity.Message;

import io.swagger.annotations.Api;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import org.springframework.data.redis.core.RedisTemplate;
@Api(value = "WebSocket服务端")
@Component
@ServerEndpoint(value = "/imserver/{userId}/{flag}")
public class WebSocketServer {
    private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class);
    private volatile static List<Session> sessions = Collections.synchronizedList(new ArrayList<>()); // 存放所有客户端的session


    // 使用 ConcurrentHashMap 代替 CopyOnWriteArraySet
    private static ConcurrentMap<String, Session> webSocketMap = new ConcurrentHashMap<>();

    // 设置为静态的 公用一个消息map ConcurrentMap为线程安全的map  HashMap不安全
    private static ConcurrentMap<String, Map<String, List<Message>>> messageMap = new ConcurrentHashMap<>(); // 消息队列


    @Autowired
    private ApplicationContext applicationContext;
    private Session session; // 与某个客户端的连接会话
    private String userId; // 用户 id

    private String readFlag="_readList";
    // 标记连接状态
    private boolean isConnected = false;
    private long lastHeartbeatTime = System.currentTimeMillis(); // 上次心跳时间

    @Autowired
    private RedisTemplate<String, String> redisTemplate = (RedisTemplate<String, String>) SpringUtil.getBean("stringRedisTemplate");

    // 心跳包相关变量（这里恢复使用，进行相应逻辑完善）
    private static final long HEARTBEAT_INTERVAL = 30; // 心跳包发送间隔，单位：秒
    private static final long HEARTBEAT_TIMEOUT = 60; // 心跳包超时时间，单位：秒
    private ScheduledExecutorService heartbeatExecutor;
    private ScheduledExecutorService heartbeatCheckExecutor; // 新增用于检查心跳包超时的定时任务线程池



    /**
     * 连接建立成功调用的方法
     *
     * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
     * @param userId  接收者用户id
     * @param flag    额外标识参数
     * @throws Exception
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId, @PathParam("flag") String flag) throws JsonProcessingException {
        this.userId = userId;
        this.session = session;
        webSocketMap.put(userId, session); // 加入到WebSocket集合中
        logger.info("新连接加入"+userId+",当前在线人数为：" + webSocketMap.size());
        isConnected = true;
        // 启动心跳包任务（完善相关逻辑）
       // startHeartbeatTask();
        // 启动心跳包超时检查任务
        //startHeartbeatCheckTask();
        // 读取离线数据
        readMessageMap(flag);
        if(!userId.contains(readFlag)){
            webSocketMap.remove(userId+readFlag);
        }
    }

    /**
     * 启动心跳包任务
     */
    private void startHeartbeatTask() {
        heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();
        heartbeatExecutor.scheduleAtFixedRate(() -> {
            if (isConnected && session.isOpen()) {
                // 添加心跳包超时判断，只有未超时才发送心跳包
                if (System.currentTimeMillis() - lastHeartbeatTime < HEARTBEAT_TIMEOUT * 1000) {
                    try {
                        this.session.getBasicRemote().sendText("heartbeat");
                        lastHeartbeatTime = System.currentTimeMillis();
//                    logger.info("发送心跳包成功，sid=" + sid);
                    } catch (IOException e) {
                        logger.error("发送心跳包失败，sid=", e);
                        handleHeartbeatFailure();
                    }
                }
            }
        }, 0, HEARTBEAT_INTERVAL, TimeUnit.SECONDS);
    }

    /**
     * 启动心跳包超时检查任务
     */
    private void startHeartbeatCheckTask() {
        heartbeatCheckExecutor = Executors.newSingleThreadScheduledExecutor();
        heartbeatCheckExecutor.scheduleAtFixedRate(() -> {
            for (Session webSocket : webSocketMap.values()) {
             //   long timeSinceLastHeartbeat = System.currentTimeMillis() - webSocket.lastHeartbeatTime;
               // if (timeSinceLastHeartbeat > HEARTBEAT_TIMEOUT * 1000) {
                    // 超时了，标记该客户端连接不正常，进行更完善的处理逻辑
                    handleHeartbeatTimeout(webSocket);
              //  }
            }
        }, 0, HEARTBEAT_INTERVAL, TimeUnit.SECONDS);
    }

    /**
     * 处理心跳包发送失败的情况（例如网络抖动等导致单次发送失败）
     */
    private void handleHeartbeatFailure() {
        int retryCount = 3; // 设置重试次数
        for (int i = 0; i < retryCount; i++) {
            try {
                Thread.sleep(1000); // 等待1秒后重试
                this.session.getBasicRemote().sendText("heartbeat");
                lastHeartbeatTime = System.currentTimeMillis();
//                logger.info("心跳包发送失败后重试成功，sid=" + sid);
                return; // 发送成功则直接返回
            } catch (IOException | InterruptedException e) {
//                logger.error("心跳包发送失败后重试第 {} 次失败，sid={}", i + 1, sid, e);
            }
        }
        // 多次重试后仍失败，标记连接不正常并关闭会话
        isConnected = false;
        try {
            session.close();
            logger.warn("多次重试后心跳包发送仍失败，关闭会话，sid=" + userId);
        } catch (IOException e) {
            logger.error("关闭会话失败，sid=" + userId, e);
        }
    }

    /**
     * 处理心跳包超时的情况（长时间未收到心跳包）
     */
    private void handleHeartbeatTimeout(Session session) {

        try {
            // 关闭对应的会话连接
            session.close();
            logger.warn("客户端心跳包超时，关闭会话，sid=" + userId);
            // 从相关集合中移除该客户端信息
            webSocketMap.remove(userId);
            // 如果有离线消息相关处理，可在这里进行（比如清理该客户端对应的离线消息缓存等）

            Map<String, List<Message>> userMessages = messageMap.get(userId);
            if (userMessages!= null) {
                userMessages.remove(userId);
                if (userMessages.isEmpty()) {
                    messageMap.remove(userId);
                }
            }
            // 减少在线人数统计
//            logger.info("心跳包超时客户端已处理，当前在线人数为：" + getOnlineCount());
            // 可以考虑在这里添加尝试重连逻辑（比如记录需要重连的客户端信息，后续定期尝试重连等）
            // 示例：重连任务可以放到一个专门的重连队列或者定时任务中去执行，这里简单打印示例
//            logger.info("准备尝试对心跳包超时客户端进行重连，sid=" + webSocket.sid);
            // 实际应用中可启动重连相关逻辑，比如调用重连方法等
        } catch (IOException e) {
            logger.error("关闭心跳包超时客户端会话失败，sid=" + userId, e);
        }
    }

    /**
     * 用户读取自己离线消息
     */
    public void readMessageMap(String flag) {
        String key = this.userId.replace(readFlag,"");
        // 处理离线消息
        Map<String, List<Message>> stringListMap = messageMap.get(key);
        if (stringListMap!= null) {
            Session receiver = webSocketMap.get(userId); // 获取接收者用户的WebSocket实例
            // 说明在用户没有登录的时候有人给用户发送消息
            List<Message> list = stringListMap.get(this.userId);
            if (flag.equals("list")) {
                Map<String, Map<String, List<Message>>> msgListMap = new HashMap<>();
                msgListMap.put("readUnLineMsg", stringListMap);
                JSONObject json = new JSONObject(msgListMap);
                try {
                        receiver.getBasicRemote().sendText(json.toString());
                     logger.info("已将离线消息展示给刚上线的用户 " + userId + ", 消息内容：");
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            } else {
                if (list!= null) {
                    for (Message msg : list) {
                        try {
                            msg.setRead(true); // 设置为已读
                            JSONObject json = new JSONObject(msg);
                            receiver.getBasicRemote().sendText(json.toString());
                            logger.info("已将离线消息发送给刚上线的用户 " + userId + ", 消息内容：" + msg.getText());
                        } catch (JsonProcessingException e) {
                            logger.error("将离线消息反序列化失败，sid=" + userId, e);
                        } catch (IOException e) {
                            logger.error("发送离线消息给刚上线的用户失败，sid=" + userId, e);
                        }
                    }
                    // 离线数据读取完成删除记录
                    stringListMap.remove(this.userId);
                    if (stringListMap.isEmpty()) {
                        messageMap.remove(this.userId);
                    }
                }
            }
        }
    }

    /**
     * 用户读取自己离线消息（另一个重载方法）
     */
    public void sendMsg2List( Session sessionList, String userId) {
        // 处理离线消息
        Map<String, List<Message>> stringListMap = messageMap.get(userId);
        if (stringListMap!= null) {
            // 说明在用户没有登录的时候有人给用户发送消息
                Map<String, Map<String, List<Message>>> msgListMap = new HashMap<>();
                msgListMap.put("readUnLineMsg", stringListMap);
                JSONObject json = new JSONObject(msgListMap);
                try {
                    sessionList.getBasicRemote().sendText(json.toString());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }

    }

    @OnClose
    public void onClose() {
          // 从map中删除
        System.out.println(this.userId + "接收者");
        webSocketMap.remove(this.userId);// 从session列表中移除
//
//        System.out.println(webSocketMap.get(this.sid) + "发送者");
        System.out.println("关闭");
        logger.info("连接关闭，sid=" + this.userId + ", 当前在线人数为："+webSocketMap.size() );
    }

    /**
     * 收到客户端消息
     *
     * @param message 消息内容
     * @throws JsonProcessingException
     */
    @OnMessage
    public void onMessage(String message) throws IOException {
        if (message.equals("heartbeat")) { // 判断是不是心跳包消息
//            logger.info("收到心跳包");
            lastHeartbeatTime = System.currentTimeMillis();
            return;
        }
          logger.info("收到消息，session=" + session + ", 消息内容：" + message);
        ObjectMapper objectMapper = new ObjectMapper();
        Message msg;
        try {
            msg = objectMapper.readValue(message, Message.class);
        } catch (JsonProcessingException e) {
            logger.error("消息反序列化失败，sid=" + this.userId + ", 消息内容：" + message, e);
            return;
        }
        String recipientID = msg.getRecipientID();
        Session session1 = webSocketMap.get(recipientID); // 获取接收者用户的WebSocket实例

        if (session1!= null &&session1.isOpen()) {
            // 接收者在线，直接发送消息
            try {
                msg.setRead(true);
                JSONObject json = new JSONObject(msg);
                session1.getBasicRemote().sendText(json.toString());
                logger.info("发送消息成功，目标客户端toUser=" + this.userId + ", 消息内容：" + msg.getText());
            } catch (IOException e) {
                logger.error("发送消息失败，目标客户端toUser=" + this.userId, e);
                logger.info("目标客户端 " + this.userId + " 在线但发送消息失败，消息已存储到Redis");
            }
        } else {
            System.out.println("接收者离线，将消息存储到messageMap" + recipientID);
            // 接收者离线，将消息存储到messageMap
            if (messageMap.get(recipientID) == null) { // 没有离线数据
                System.out.println("接收者离线，之前没有消息，创建新的map" + recipientID);
                Map<String, List<Message>> maps = new HashMap<>();
                System.out.println("接收者离线，之前没有消息，创建新的list" + recipientID);
                List<Message> list = new ArrayList<>();
                list.add(msg);
                maps.put(recipientID, list);
                messageMap.put(recipientID, maps);
            } else { // 已经有未读数据
                Map<String, List<Message>> listObject = messageMap.get(recipientID);
                List<Message> objects = listObject.get(recipientID);
                if (objects == null) {
                    objects = new ArrayList<>();
                }
                objects.add(msg);
                listObject.put(recipientID, objects);
                messageMap.put(recipientID, listObject);
            }
            // 将消息发送个列表界面
            String receiverKey = recipientID + readFlag;
            Session sessionList = webSocketMap.get(receiverKey);
            if (sessionList!= null && sessionList.isOpen()) {
                sendMsg2List(sessionList, recipientID);
            }
        }
    }

    @OnError
    public void onError(Session session, Throwable error) {
        logger.error("发生错误", error);
    }



}